package ex.other;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * demo:
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/connectors/datastream/jdbc/
 * apidoc:
 * https://nightlies.apache.org/flink/flink-docs-release-1.17/api/java/
 * 
 * public static <T> SinkFunction<T> sink(String sql,
                                       JdbcStatementBuilder<T> statementBuilder,//本例中使用lambda，看的不清楚，
                                       JdbcExecutionOptions executionOptions,
                                       JdbcConnectionOptions connectionOptions)
 * @author Administrator
 */
public class SinkDemo {
    
    static class Book {
        public Book(Integer id, String title, String authors, Integer year) {
            this.id = id;
            this.title = title;
            this.authors = authors;
            this.year = year;
        }
        final Integer id;
        final String title;
        final String authors;
        final Integer year;
    }
    
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
//        MyJdbcStatementBuilder jdbcStatementBuiler = new MyJdbcStatementBuilder();
        env.fromElements(
                new Book(101, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki Kalavri", 2019),
                new Book(102, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax", 2018),
                new Book(103, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
                new Book(104, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd Palino", 2017)
        ).addSink(
                JdbcSink.sink(
                        "insert into book (id, title, authors, year) values (?, ?, ?, ?)",
                        (statement, book) -> {
                            statement.setLong(1, book.id);
                            statement.setString(2, book.title);
                            statement.setString(3, book.authors);
                            statement.setInt(4, book.year);
                        },
                        JdbcExecutionOptions.builder()
                                .withBatchSize(1000)
                                .withBatchIntervalMs(200)
                                .withMaxRetries(5)
                                .build(),
                         new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()

                                .withDriverName("com.mysql.cj.jdbc.Driver")
                                .withUsername("root")
                                .withPassword("liji@2024")
                                .withUrl("jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=Asia/Shanghai&useSSL=false")
                                .build()
                ));
                
        env.execute();
    }
}
